package com.spring.nio.service.file.task;

import com.spring.nio.common.BasicConstant;
import com.spring.nio.core.event.listener.event.constant.ExceptionEventTypeEnum;
import com.spring.nio.core.event.listener.event.model.ExceptionEventSource;
import com.spring.nio.core.event.listener.event.model.FileHandlerExceptionEventSource;
import com.spring.nio.core.exception.ExceptionManager;
import com.spring.nio.service.file.frame.FileMessageFrame;
import com.spring.nio.service.file.model.DownloadResult;
import com.spring.nio.service.file.model.structure.FileTransportTaskDTO;
import com.spring.nio.service.protocol.response.file.FileTransportResponse;
import com.spring.nio.util.BasicUtil;
import com.spring.nio.util.LocalTime;
import com.spring.nio.util.NioUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * 文件下载传输线程
 * @author spring
 * @description 文件下载线程: 从代码角度讲，影响下载快慢的因素为，文件流包大小以及发送速度
 *  文件下载线程中的文件通道关闭和传输通道关闭由统一的NioUtil.closedAndRelease()方法进行释放
 */
@Slf4j
public class FileDownloadStreamCallable implements Callable<DownloadResult> {
    private volatile Boolean stop = Boolean.FALSE;
    private FileTransportTaskDTO fileTransportTaskDto;

    public FileDownloadStreamCallable(FileTransportTaskDTO fileTransportTaskDto) {
        this.fileTransportTaskDto = fileTransportTaskDto;
    }

    @Override
    public DownloadResult call() {
        try {
            if(!this.fileTransportTaskDto.getFileChannel().isOpen()) {
                return DownloadResult.generate(BasicConstant.FILE_CHANNEL_OPEN_ERROR, Boolean.FALSE);
            }

            // 该方法一旦异常，执行异常处理，将结束FileDownloadStreamCallable并返回任务结果，另外当前任务执行异常处理，设置任务状态为EXCEPTION，
            return this.loopSendFileDownloadStream();
        } catch (Exception e) {
            // 异常事件(ExceptionEvent) 事件源类型为异常事件源类型(EXCEPTION_EVENT_SOURCE)  事件源数据(NioReadEventExceptionSource)
            FileHandlerExceptionEventSource fileHandlerExceptionEventSource = new FileHandlerExceptionEventSource();
            fileHandlerExceptionEventSource.setSocketChannelContext(this.fileTransportTaskDto.getSocketChannelContext());
            fileHandlerExceptionEventSource.setFileTransportTaskDto(this.fileTransportTaskDto);
            fileHandlerExceptionEventSource.setModule("FileDownloadStreamCallable --> method [call] --> 文件下载流帧处理异常");
            fileHandlerExceptionEventSource.setException(e);
            fileHandlerExceptionEventSource.setExceptionEventTypeEnum(ExceptionEventTypeEnum.FILE_TRANSPORT_HANDLER_IO_EXCEPTION);
            fileHandlerExceptionEventSource.setIsSend(Boolean.TRUE);
            ExceptionManager.handleException(fileHandlerExceptionEventSource);
            return DownloadResult.generate("", Boolean.FALSE, e.getCause().getMessage());
        }
    }

    /**
     * 循环发送文件流数据
     * @return
     *
     *         // 4B -> 当前帧的总长度；
     *         // 1B -> 文件是否读取完[0:未读取完, 1:读取完],
     *         // 4B -> 包序号
     *         // 4B -> 文件字节流长度
     */
    private DownloadResult loopSendFileDownloadStream() throws IOException, InterruptedException {
        int originStreamDataIndex = 4 + 1 + 4 + 4;
        // 包序号
        Integer packetIndex = 0;
        // 设置任务为处理中
        this.fileTransportTaskDto.setTaskStatus("HANDLING");
        ByteBuffer byteBuffer = this.fileTransportTaskDto.getSocketChannelContext().getByteBuffer();
        int readFileSize = 0;
        Long sumReadFileSize = 0L;

        // 2、计算当前待下载文件每次发送的字节量，根据当前服务端正在进行的文件下载任务数，以及当前服务端可用内存大小动态配置当前待下载文件的字节发送量
        // 使用stoped来保证线程是否需要被终止，如果在while后stoped被置为true，那么无需执行其他操作，线程可直接结束
        int i = 0;
        while ((readFileSize = this.fileTransportTaskDto.getFileChannel().read(byteBuffer)) > 0 && !this.getStop()) {
            // 统计下载字节数
            sumReadFileSize += readFileSize;

            // 待发送的字节数组
            byte[] sendBytes = new byte[originStreamDataIndex + readFileSize];
            // 封装文件发送流
            this.encapsulationStreamData(sumReadFileSize, this.fileTransportTaskDto.getFileLength(), packetIndex++, byteBuffer, sendBytes);
            // 包装文件流
            FileTransportResponse fileTransportResponse = new FileTransportResponse();
            fileTransportResponse.setCode("200");
            fileTransportResponse.setFileName(this.fileTransportTaskDto.getFile().getName());
            fileTransportResponse.setFrameType(FileMessageFrame.FrameType.DATA_TRANSPORT.getBit());
            fileTransportResponse.setStreamData(sendBytes);
            fileTransportResponse.setTime(Timestamp.valueOf(LocalDateTime.now()));

            /**
             * 睡眠时间: 下载文件时，由于通过异步队列方式将文件流数据下发给客户端，存在网络发送缓慢，文件流入队速度过快，导致队列很快塞满很多数据被丢弃从而导致数据发送不完整
             *  此时需要控制入队的速度，数据出队是通过nio WriteEvent触发，只要发送缓冲区不满，该事件就会一直触发，但是存在多个通道都订阅了写事件，每个通道对应的
             *  写事件被触发存在时间延迟，导致正在下载的文件通道迟迟不能触发写事件，从而队列中的数据无法被出队发送，那么队列很快就会溢出，浪费大量内存
             *
             *  加入睡眠时长，来控制下载数据入队速度，进而保证nio写事件消费端能够快速将队列中的数据订阅并发送出去
             * */
            TimeUnit.MILLISECONDS.sleep(300);
            this.fileTransportTaskDto.getSocketChannelContext().getBlockingQueue().add(fileTransportResponse);
            //log.info("[ " + LocalTime.formatDate(LocalDateTime.now()) + " ] FileDownloadStreamCallable | --> 下载文件 = [{}] 已读取和发送第 [{}] 个包数据入队成功，当前通道队列待发送数据个数 = [{}]", this.fileTransportTaskDto.getFile().getName(), packetIndex, this.fileTransportTaskDto.getSocketChannelContext().getBlockingQueue().size());
            /*if(i++ == 500) {
                throw new IOException("文件流发送异常");
            }*/
        }

        if(readFileSize == -1) {
            log.info("[ " + LocalTime.formatDate(LocalDateTime.now()) + " ] FileDownloadStreamCallable | --> 下载文件 = [{}] 已读取完成(读取最后结果为[{}])", this.fileTransportTaskDto.getFile().getName(), readFileSize);
            // 文件流读取完成，返回文件下载结果
            return DownloadResult.generate("文件 [" + this.fileTransportTaskDto.getFile().getName() + "] 下载完成", Boolean.TRUE);
        }

        log.info("[ " + LocalTime.formatDate(LocalDateTime.now()) + " ] FileDownloadStreamCallable | --> 下载文件 = [{}] 任务被取消", this.fileTransportTaskDto.getFile().getName(), readFileSize);
        return DownloadResult.generate("文件 = [" + this.fileTransportTaskDto.getFile().getName() + "] 下载任务取消", Boolean.FALSE);
    }

    /**
     * 封装文件发送流
     * @param sumReadFileSize 当前从文件读取到的字节总数
     * @param fileSize   当前文件大小
     * @param packetIndex 当前文件字节数据包序号
     * @param byteBuffer 文件文件字节数据
     * @param sendBytes  待发送的字节数组
     */
    private void encapsulationStreamData(Long sumReadFileSize, Long fileSize, Integer packetIndex, ByteBuffer byteBuffer, byte[] sendBytes) {

        // 1、当前帧总长度 4B (0~3)
        byte[] frameSumLength = BasicUtil.intToBytes(sendBytes.length);
        System.arraycopy(frameSumLength, 0, sendBytes, 0, frameSumLength.length);

        // 2、是否是结束帧 1B (4)
        sendBytes[4] = (sumReadFileSize.equals(fileSize))?(byte)1:(byte)0;

        // 3、包序号 4B (5~8)
        byte[] packetIndexBytes = BasicUtil.intToBytes(packetIndex);
        System.arraycopy(packetIndexBytes, 0, sendBytes, 5, packetIndexBytes.length);

        // 4、文件流数据
        byteBuffer.flip();
        while(byteBuffer.hasRemaining()) {
            byte[] fileStreamBytes = new byte[byteBuffer.remaining()];
            byteBuffer.get(fileStreamBytes);

            // 5、文件字节流长度 4B (9 ~ 12)
            byte[] fileStreamByteLength = BasicUtil.intToBytes(fileStreamBytes.length);
            System.arraycopy(fileStreamByteLength, 0, sendBytes, 9, fileStreamByteLength.length);

            // 6、拷贝文件流数据
            System.arraycopy(fileStreamBytes, 0, sendBytes, 13, fileStreamBytes.length);
        }

        byteBuffer.clear();
    }

    public Boolean getStop() {
        return stop;
    }

    public void setStop(Boolean stop) {
        this.stop = stop;
    }
}
